package com.yanqu.road.task;

import com.yanqu.road.message.YanQuMessage;
import com.yanqu.road.message.YanQuMessageBuffer;
import com.yanqu.road.connect.AgentServerConnection;
import io.netty.channel.ChannelPromise;

public class FlushAgentServerConnectionTask implements Runnable{
    private AgentServerConnection agentServerConnection;

    public FlushAgentServerConnectionTask(AgentServerConnection agentServerConnection) {
        this.agentServerConnection = agentServerConnection;
    }

    @Override
    public void run() {
        agentServerConnection.getFlushSwitch().set(true);
        if (agentServerConnection.getMessageQueue().size() == 1) {
            YanQuMessage msg = agentServerConnection.getMessageQueue().poll();
            if (msg != null) {
                agentServerConnection.getChannel().writeAndFlush(msg, newPromise());
            }
        }else if (agentServerConnection.getMessageQueue().size() > 1) {
            YanQuMessageBuffer messageBuffer = new YanQuMessageBuffer();
            agentServerConnection.getMessageQueue().drainTo(messageBuffer);
            if (messageBuffer.size() > 0) {
                agentServerConnection.getChannel().writeAndFlush(new YanQuMessage(messageBuffer), newPromise());
            }
        }
    }

    private ChannelPromise newPromise() {
        ChannelPromise promise = agentServerConnection.getChannel().newPromise();
        promise.addListener(agentServerConnection.getListener());
        agentServerConnection.getCounter().incrementAndGet();
        return promise;
    }
}
